with tab_centerarr as (
    SELECT
        waybill_no ,
        scan_time ,
        site_agent_code ,
        site_agent_name ,
        scan_site_code ,
        scan_site
    from (
             select
                 waybill_no ,
                 scan_time ,
                 site_agent_code ,
                 site_agent_name ,
                 scan_site_code ,
                 scan_site ,
                 row_number() over(partition by waybill_no order by scan_time desc) as rk
             from jms_dwd.dwd_tab_barscan_centerarrival_base_hi
             where dt between date_add('{{ execution_date | cst_ds }}',-9) and date_add('{{ execution_date | cst_ds }}',-0)
               and input_time between concat(date_add('{{ execution_date | cst_ds }}',-8),' 07:00:00')
                              and concat(date_add('{{ execution_date | cst_ds }}',-1),' 07:00:00')
               and substr(waybill_no,1,1)!='B'
         )t where rk=1
),
--是否有其他操作
tab_other_op as
    (
        select
            waybill_no ,scan_type ,scan_time,network_code
        from
            (
                select
                    waybill_no ,scan_type ,scan_time,network_code,
                    row_number() over(partition by waybill_no order by scan_time desc) as rk
                from
                    (
                        select waybill_no ,scan_type ,scan_time,scan_site_code as network_code
                        from jms_dwd.dwd_tab_barscan_arrival_base_hi
                        where substr(dt,1,10) between date_add('{{ execution_date | cst_ds }}',-9) and date_add('{{ execution_date | cst_ds }}',1)
                          and dt <= concat(date_add('{{ execution_date | cst_ds }}',1),'-05')
                        union all
                        select waybill_no ,scan_type ,scan_time,scan_site_code as network_code
                        from  jms_dwd.dwd_tab_barscan_send_base_hi
                        where substr(dt,1,10) between date_add('{{ execution_date | cst_ds }}',-9) and date_add('{{ execution_date | cst_ds }}',1)
                          and dt <= concat(date_add('{{ execution_date | cst_ds }}',1),'-05')
                        union all
                        select waybill_no ,scan_type ,scan_time,scan_site_code as network_code
                        from  jms_dwd.dwd_tab_barscan_deliver_base_hi
                        where substr(dt,1,10) between date_add('{{ execution_date | cst_ds }}',-9) and date_add('{{ execution_date | cst_ds }}',1)
                          and dt <= concat(date_add('{{ execution_date | cst_ds }}',1),'-05')
                        union all
                        select waybill_no ,scan_type ,scan_time,scan_site_code as network_code
                        from  jms_dwd.dwd_tab_barscan_collect_base_hi
                        where substr(dt,1,10) between date_add('{{ execution_date | cst_ds }}',-9) and date_add('{{ execution_date | cst_ds }}',1)
                          and dt <= concat(date_add('{{ execution_date | cst_ds }}',1),'-05')
                        union all
                        select waybill_no ,scan_type ,signed_time as scan_time,signed_site as network_code
                        from jms_dwd.dwd_tab_barscan_sign_base_hi
                        where substr(dt,1,10) between date_add('{{ execution_date | cst_ds }}',-9) and date_add('{{ execution_date | cst_ds }}',1)
                          and dt <= concat(date_add('{{ execution_date | cst_ds }}',1),'-05')
                    ) tt
            )t where rk=1
    ),
--是否有其他操作
tab_other2 as
    (
        select
            waybill_no ,scan_type ,scan_time,network_code
        from
            (
                select
                    waybill_no ,scan_type ,scan_time,network_code,
                    row_number() over(partition by waybill_no order by scan_time ) as rk
                from
                    (
                        select waybill_no ,scan_type ,scan_time,scan_site_code as network_code
                        from  jms_dwd.dwd_tab_barscan_collect_base_hi
                        where substr(dt,1,10) between date_add('{{ execution_date | cst_ds }}',-30) and date_add('{{ execution_date | cst_ds }}',-2)
                          and dt <= concat(date_add('{{ execution_date | cst_ds }}',2),'-07')
                          and scan_type ='出库扫描'
                        union all
                        select waybill_no ,scan_type ,signed_time as scan_time,signed_site as network_code
                        from jms_dwd.dwd_tab_barscan_sign_base_hi
                        where substr(dt,1,10) between date_add('{{ execution_date | cst_ds }}',-30) and date_add('{{ execution_date | cst_ds }}',-2)
                          and dt <= concat(date_add('{{ execution_date | cst_ds }}',2),'-07')
                    ) tt
            )t where rk=1
    ),
--运单派件网点
tab_disp as (
    select * from (
                      select
                          waybill_no , --运单号
                          first_code,
                          second_code,
                          network_code,
                          provider_desc,
                          city_desc,
                          area_desc,
                          quanzhong,
                          row_number() over(partition by waybill_no order by quanzhong desc ) as ranks
                      from
                          (
                              select *from (
                                               select t1.waybill_no,                                                    --运单号
                                                      split(t1.new_terminal_dispatch_code, ' ')[0] as first_code,    --一段码
                                                      split(t1.new_terminal_dispatch_code, ' ')[1] as second_code,    --一段码
                                                      t2.network_code,
                                                      row_number() over(partition by t1.waybill_no order by print_time desc) as rk,
                                                       10 as quanzhong
                                               from jms_dwd.dwd_tab_reback_transfer_express_base_hi t1
                                                        left join jms_dim.dim_first_second_third_code_info t2 on split(t1.new_terminal_dispatch_code, ' ')[0] = t2.first_code and split(t1.new_terminal_dispatch_code, ' ')[1]=t2.second_code
                                               where
                                                       substr(dt,1,10) >= date_add('{{ execution_date | cst_ds }}', -30)
                                                 and t1.print_flag='1' --面单已经打印
                                           )t where rk=1
                              union ALL
                              select t1.waybill_no , --运单号
                                     t1.first_code,
                                     t1.second_code,
                                     t2.network_code,
                                     10 as rk,
                                     1 as quanzhong
                              from jms_dwd.dwd_yl_oms_oms_waybill_incre_hi t1
                                       left join jms_dim.dim_first_second_third_code_info t2 on t1.first_code = t2.first_code and t1.second_code=t2.second_code
                              where substr(dt, 1, 10) BETWEEN date_add ('{{ execution_date | cst_ds }}', -45) and date_add('{{ execution_date | cst_ds }}',1)
                          ) t
                              left join jms_dim.dim_network_whole_massage net1 on net1.code = t.network_code
                  ) tt
    where tt.ranks=1
),
--问题件登记表
tab_problem as
    (
        select
            registration_network_code --登记网点
             , second_level_type_code
             , second_level_type_name
             , waybill_no
             , quanzhong
        from
            (
                select
                    registration_network_code --登记网点
                     , second_level_type_code
                     , second_level_type_name
                     , waybill_no
                     , if(second_level_type_code ='66e' or second_level_type_code ='66g' ,10,1)  as  quanzhong--问题件登记
                     , row_number() over(partition by waybill_no,registration_network_code order by if(second_level_type_code ='66e' or second_level_type_code ='66g' ,10,1) desc ) as rk
                from
                    jms_dwd.dwd_sqs_registration_problem_piece_base_dt
                where substr(dt,1,10) >= date_add('{{ execution_date | cst_ds }}', -30)
            )t where rk=1
    ),
interceptor_network as (--拦截表 --网点
    select network_code,
           if(effect_time_type = 1,effect_time_start,'2022-01-01') as effect_time_start,
           if(effect_time_type = 1,effect_time_end,'9999-01-01')  as effect_time_end
    from jms_dim.dim_yl_oms_order_interceptor_hi
    where data_type = 'network'
      and is_delete = 0 --未删除
      and is_enable = 1 --已启用
),
interceptor_area_province as (--拦截表 --区域
    select province, null as city, null as area,
           if(effect_time_type = 1,effect_time_start,'2022-01-01') as effect_time_start,
           if(effect_time_type = 1,effect_time_end,'9999-01-01')  as effect_time_end
    from jms_dim.dim_yl_oms_order_interceptor_hi
    where is_delete = 0
      and is_enable = 1
      and data_type = 'area'
      and province is not null
      and city is null
),
interceptor_area_city as (--拦截表 --区域
    select province, city, null as area,
           if(effect_time_type = 1,effect_time_start,'2022-01-01') as effect_time_start,
           if(effect_time_type = 1,effect_time_end,'9999-01-01')  as effect_time_end
    from jms_dim.dim_yl_oms_order_interceptor_hi
    where is_delete = 0
      and is_enable = 1
      and data_type = 'area'
      and province is not null
      and city is not null
      and area is null
),
interceptor_area_area as (--拦截表 --区域
    select province, city, area,
           if(effect_time_type = 1,effect_time_start,'2022-01-01') as effect_time_start,
           if(effect_time_type = 1,effect_time_end,'9999-01-01')  as effect_time_end
    from jms_dim.dim_yl_oms_order_interceptor_hi
    where is_delete = 0
      and is_enable = 1
      and data_type = 'area'
      and province is not null
      and city is not null
      and area is not null
)

insert overwrite table jms_dm.dm_center_detention_detail_dt partition(dt)
select
    main.waybill_no ,
    main.scan_time ,
    main.site_agent_code ,
    main.site_agent_name ,
    main.scan_site_code ,
    main.scan_site,
    t1.first_code,
    t1.second_code,
    t1.network_code as dispatch_code, --派件网点
    t1.provider_desc,
    t1.city_desc,
    t1.area_desc,
    if(t1.quanzhong=10,1,0) as  is_reback,
    if(t2.quanzhong=10,1,0) as  is_yiqing_problem ,--是否是疫情件
    --case when t2.quanzhong=10 then 1 --是疫情件
    --     when t2.quanzhong=1  then 2 --是问题件非疫情件
    --     else null  --非问题件
    --     end as is_yiqing_problem,
    t2.second_level_type_code, --问题件类型code
    t2.second_level_type_name,  --问题件类型名称
    case when p.province is not null then 1
         when c.city is not null then 1
         when a.area is not null then 1
         when n.network_code is not null then 1
         else 0
        end as is_intercepte, --是否是拦截件
    --from_unixtime(unix_timestamp() ,'yyyy-MM-dd HH:mm:ss') as sys_time,
    concat(date_add('{{ execution_date | cst_ds }}',1),' 05:00:00') as sys_time,
    (unix_timestamp(concat(date_add('{{ execution_date | cst_ds }}',1),' 05:00:00'))-unix_timestamp(main.scan_time))/(60*60) as hours,
    case when (unix_timestamp(concat(date_add('{{ execution_date | cst_ds }}',1),' 05:00:00'))-unix_timestamp(main.scan_time))/(60*60) >=48  and (unix_timestamp(concat(date_add('{{ execution_date | cst_ds }}',1),' 05:00:00'))-unix_timestamp(main.scan_time))/(60*60) < 72 then '48h'
         when (unix_timestamp(concat(date_add('{{ execution_date | cst_ds }}',1),' 05:00:00'))-unix_timestamp(main.scan_time))/(60*60) >=72  and (unix_timestamp(concat(date_add('{{ execution_date | cst_ds }}',1),' 05:00:00'))-unix_timestamp(main.scan_time))/(60*60) < 96 then '72h'
         when (unix_timestamp(concat(date_add('{{ execution_date | cst_ds }}',1),' 05:00:00'))-unix_timestamp(main.scan_time))/(60*60) >=96  and (unix_timestamp(concat(date_add('{{ execution_date | cst_ds }}',1),' 05:00:00'))-unix_timestamp(main.scan_time))/(60*60) < 120 then '96h'
         when (unix_timestamp(concat(date_add('{{ execution_date | cst_ds }}',1),' 05:00:00'))-unix_timestamp(main.scan_time))/(60*60) >=120 and (unix_timestamp(concat(date_add('{{ execution_date | cst_ds }}',1),' 05:00:00'))-unix_timestamp(main.scan_time))/(60*60) < 144 then '120h'
         when (unix_timestamp(concat(date_add('{{ execution_date | cst_ds }}',1),' 05:00:00'))-unix_timestamp(main.scan_time))/(60*60) >=144 and (unix_timestamp(concat(date_add('{{ execution_date | cst_ds }}',1),' 05:00:00'))-unix_timestamp(main.scan_time))/(60*60) < 168 then '144h'
         when (unix_timestamp(concat(date_add('{{ execution_date | cst_ds }}',1),' 05:00:00'))-unix_timestamp(main.scan_time))/(60*60) >=168  then '168h'
         else  '小于48h' end as hour_type,
    main.regional_desc,
    '{{ execution_date | cst_ds }}' as  date_time,
    --date(scan_time) as dt
    '{{ execution_date | cst_ds }}' as dt
from
    (
        select
        main.waybill_no ,
        main.scan_time ,
        net1.agent_code as site_agent_code ,
        net1.agent_name as site_agent_name ,
        main.scan_site_code ,
        main.scan_site,
        net1.regional_desc
        from tab_centerarr  main
        left join tab_other_op t2 on  main.waybill_no =t2.waybill_no  and  main.scan_time < t2.scan_time
        left join tab_other2 t3 on  main.waybill_no =t3.waybill_no  and  main.scan_time > t3.scan_time
        left join  jms_dim.dim_network_whole_massage net1 on net1.code = main.scan_site_code
        where t2.waybill_no is null and t3.waybill_no  is  null
    ) main
    left join  tab_disp t1 on  main.waybill_no=t1.waybill_no
    left join tab_problem t2 on t2.waybill_no=main.waybill_no and  t2.registration_network_code=main.scan_site_code
    left join interceptor_area_province p on t1.provider_desc=p.province and main.scan_time >=p.effect_time_start and main.scan_time <=p.effect_time_end
    left join interceptor_area_city c on t1.city_desc=c.city and t1.provider_desc=c.province and main.scan_time >=c.effect_time_start and main.scan_time <=c.effect_time_end
    left join interceptor_area_area a on t1.area_desc=a.area and t1.provider_desc=a.province and t1.city_desc=a.city and main.scan_time >=a.effect_time_start and main.scan_time <=a.effect_time_end
    left join interceptor_network n on t1.network_code=n.network_code and main.scan_time >=n.effect_time_start and main.scan_time <=n.effect_time_end
